package cn.web03.websocket.send;

import cn.web03.websocket.po.SocketMessage;
import cn.web03.websocket.po.SocketMessageT;
import cn.web03.websocket.po.SocketUser;
import cn.web03.websocket.service.SocketMessageService;
import cn.web03.websocket.service.SocketUserService;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.springframework.stereotype.Component;

import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.Future;
/**
 * @author 零三 2020.06
 * https://web03.cn
 */
/**
 *
 * 用户进入系统->输入信息->连接socket->将信息存入数据库->
 * 获取用户id，查询
 * socketUser.id = launchId,取receiveUser为arr1
 * socketUser.id = receiveId，取launchUser为arr2
 * 合并arr1和arr2，去重，返回前端
 *
 *
 **/
@ServerEndpoint("/chat/room/group/{param}")
@Component
public class MyWebsocketServer {
    // 正确引用
    public static SocketUserService socketUserService;
    public static SocketMessageService socketMessageService;

    // 记录空闲Session集合
    private static CopyOnWriteArraySet<Session> idle = new CopyOnWriteArraySet<Session>();
    // 记录正在使用中Session集合，value为Future，表示使用情况
    private static ConcurrentHashMap<Session, Future<Void>> busy = new ConcurrentHashMap<Session, Future<Void>>();
    /**
     * 存放所有在线的客户端以及客户端信息
     */
    private static Map<String, Session> clientSession = new ConcurrentHashMap<>();
    private static Map<String, SocketUser> clientUser = new ConcurrentHashMap<>();

    @OnOpen
    public void onOpen(Session session,@PathParam("param")String socketUserId) {
         System.out.println("有新的客户端连接了: "+session.getId());
        SocketUser socketUser = socketUserService.getSocketUser(Long.valueOf(socketUserId));
        socketUser.setSocketId(session.getId());
        //更新当前用户的连接id存入数据库
        socketUserService.saveSocketUsers(socketUser);
        socketUser.setPassword(null);
        //将新用户存入在线的组
        clientUser.put(session.getId(), socketUser);
        //将客户端存入
        clientSession.put(session.getId(),session);

        //判断是否在线
        for (Map.Entry<String, SocketUser> user : clientUser.entrySet()) {
            SocketUser socketUser1 = user.getValue();
             System.out.println("在线--"+socketUser1);
            if (socketUser1.getId() == socketUser.getId()){
                SocketMessage socketMessage = new SocketMessage(null,"0",null,"0",null, "已在其他设备登陆，强制下线",5,new Date());
                Session session1 = clientSession.get(socketUser1.getSocketId());
                sendTo(socketMessage,session1);
                break;
            }
        }
        open(session);
        SocketMessage socketMessage = new SocketMessage(null,String.valueOf(socketUser.getId()),null,"0",null,socketUser.getNickname()  + "进入聊天室",0,new Date());
        this.sendAll(socketMessage);
        this.userAll();
        this.inLine();
    }

    /**
     * 客户端关闭
     * @param session session
     */
    @OnClose
    public void onClose(Session session) {
         System.out.println("用户" + session.getId() + "离线");
        SocketUser socketUser = clientUser.get(session.getId());

        //将掉线的用户移除在线的组里
        clientSession.remove(session.getId());
        clientUser.remove(session.getId());
        close(session);
        SocketMessage socketMessage = new SocketMessage(null,"0",null,"0",null,socketUser.getNickname()  + "离开聊天室",0,new Date());
        this.sendAll(socketMessage);
        this.inLine();
    }

    /**
     * 发生错误
     * @param throwable e
     */
    @OnError
    public void onError(Throwable throwable) {
        throwable.printStackTrace();
    }


    /**
     * 收到客户端发来消息
     * @param message  消息对象
     */
    @OnMessage
    public void onMessage(String message) {
         System.out.println("收到消息：" + message);
        JSONObject jsStr = JSONObject.parseObject(message);
        SocketMessage socketMessage = JSONObject.toJavaObject(jsStr, SocketMessage.class);
        socketMessage.setCreateTime(new Date());

        SocketUser socketUser = clientUser.get(socketMessage.getLaunchId());
        socketMessage.setLaunchUser(socketUser);
        socketMessage.setLaunchId(String.valueOf(socketUser.getId()));

         if (!"-1".equals(socketMessage.getReceiveId())){
             //离线消息
             if (socketMessage.getType() == 30 || socketMessage.getType() == 40){
                 //将客户端提交的type设置为服务端的type
                 if (socketMessage.getType() == 40)
                     socketMessage.setType(4);
                 if (socketMessage.getType() == 30)
                     socketMessage.setType(3);
                 //数据库里边查找用户信息
                 SocketUser receiveUser = socketUserService.getSocketUser(Long.valueOf(socketMessage.getReceiveId()));
                 socketMessage.setReceiveUser(receiveUser);
                 //保存到数据库
                 saveMessage(socketMessage);
             } else {
                 SocketUser receiveUser = clientUser.get(socketMessage.getReceiveId());
                 Session receiveSession = clientSession.get(socketMessage.getReceiveId());//接收人
                 socketMessage.setReceiveUser(receiveUser);
                 socketMessage.setReceiveId(String.valueOf(receiveUser.getId()));
                 this.sendTo(socketMessage,receiveSession);
             }
         } else {
             socketMessage.setReceiveId("0");
             this.sendAll(socketMessage);
         }

    }


    /**
     * 群发消息
     * @param socketMessage 消息内容
     */
    private void sendAll(SocketMessage socketMessage) {
//        for (Map.Entry<String, Session> sessionEntry : clientSession.entrySet()) {
//            if (sessionEntry.getValue().isOpen())
//            sessionEntry.getValue().getAsyncRemote().sendText(JSON.toJSONString(socketMessage));
//        }
        for (Map.Entry<String, Session> sessionEntry : clientSession.entrySet()) {
            try {
                send(sessionEntry.getValue(), socketMessage, 3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        saveMessage(socketMessage);
    }

    /**
     * 发送给指定用户
     * @param socketMessage session
     * @param receiveSession session
     */
    private void sendTo(SocketMessage socketMessage,Session receiveSession) {
        saveMessage(socketMessage);
        if (receiveSession != null) {
            try {
                send(receiveSession, socketMessage, 3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
//            try {
//                receiveSession.getBasicRemote().sendText(JSON.toJSONString(socketMessage));
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
        }
    }


    /**
     * 将消息保存在数据库
     * @param socketMessage
     */
    private void saveMessage(SocketMessage socketMessage){
        if (socketMessage.getType() != 1 && socketMessage.getType() != 2&& socketMessage.getType() != 5){
            SocketMessageT socketMessageT = new SocketMessageT(null,socketMessage.getLaunchId(),JSON.toJSONString(socketMessage.getLaunchUser()),socketMessage.getReceiveId(),JSON.toJSONString(socketMessage.getReceiveUser()),socketMessage.getContent(),socketMessage.getType(),socketMessage.getCreateTime());
            socketMessageService.save(socketMessageT);
        }
    }
    /**
     * 获取在线人数
     */
    private void inLine(){
        SocketMessage socketMessage = new SocketMessage(null,"0",null,"0",null,JSON.toJSONString(clientUser),1,new Date());
        this.sendAll(socketMessage);
    }
    /**
     * 获取总人数
     */
    private void userAll(){
        List<SocketUser> socketUsers = socketUserService.getAllSocketUser();
        List<SocketUser> socketUsers1 = new ArrayList<>();
        for (SocketUser socketUser: socketUsers) {
            socketUser.setPassword(null);
            socketUsers1.add(socketUser);
        }
        SocketMessage socketMessage = new SocketMessage(null,"0",null,"0",null,JSON.toJSONString(socketUsers1),2,new Date());
        this.sendAll(socketMessage);
    }




    // 新增Session
    public static void open(Session session) {
        idle.add(session);
    }

    // 关闭Session
    public static void close(Session session) {
        idle.remove(session);
        busy.remove(session);
    }

    // 使用session发送消息
    public static void send(Session session, SocketMessage message, Integer timeout) throws InterruptedException {
        if (timeout < 0) { // timeout后放弃本次发送
            return;
        }
        if (idle.remove(session)) { // 判断session是否空闲，抢占式
           busy.put(session, session.getAsyncRemote().sendText(JSON.toJSONString(message)));
        } else {
            // 若session当前不在idle集合，则去busy集合中查看session上次是否已经发送完毕，即采用惰性判断
            synchronized (busy) {
                if (busy.containsKey(session) && busy.get(session).isDone()) {
                    busy.remove(session);
                    idle.add(session);
                }
            }
            // 重试
            Thread.sleep(100);
            send(session, message, timeout - 100);
        }
    }

}
